package edu.ucla.cs.rpc.multicast.privilege;

import java.io.IOException;
import java.net.SocketAddress;
import java.util.concurrent.locks.Condition;

import edu.ucla.cs.rpc.multicast.Sender;
import edu.ucla.cs.rpc.multicast.network.MulticastMessageSender;
import edu.ucla.cs.rpc.multicast.network.message.Message;
import edu.ucla.cs.rpc.multicast.util.token.NetworkTokenManager;
import edu.ucla.cs.rpc.multicast.util.token.Token;
import edu.ucla.cs.rpc.multicast.util.token.TokenNotHeldException;

/**
 * A PrivilegedSender is a Sender that delays multi-casting messages until it
 * holds a permission Token. The Token is circulated amongst a group of
 * PrivilegedSenders.
 * 
 * @author Philip Russell
 * @see Total Order Broadcast and Multicast Algorithms: Taxonomy and Survery,
 *      Defago et al
 * @see Section 4.3, pages 14 - 16
 * 
 */
public class PrivilegedSender implements Sender {

	/* The token communications agent */
	private NetworkTokenManager tokenManager = null;

	/* The communications agent for sending messages to Destinations */
	private MulticastMessageSender sender = null;

	private PrivilegedSender() {
	}

	/**
	 * Primary constructor. Creates a PrivilegedSender that manages sending
	 * messages via the multicastManager, and listens for the token on the
	 * tokenAddress.
	 * 
	 * @param multicastManager
	 *            Multicast communications agent for sending messages.
	 * @param tokenAddress
	 *            Address to listen for the token.
	 * @throws IOException
	 *             Thrown if a lower-level network error is encountered, such as
	 *             failing to create/bind a socket.
	 */
	public PrivilegedSender(SocketAddress multicastManager,
			SocketAddress tokenAddress) throws IOException {

		this.tokenManager = new NetworkTokenManager(tokenAddress);
		this.sender = new MulticastMessageSender(multicastManager);
	}

	/**
	 * Sets the token, needed to initialize the token ring of PrivilegedSenders.
	 * 
	 * @param token
	 *            The token.
	 */
	public void setToken(Token token) {
		this.tokenManager.setToken(token);
	}

	/**
	 * 
	 * @return The address that this PrivilegedSender listens for the token on.
	 */
	public SocketAddress getTokenAddress() {
		return this.tokenManager.getAddress();
	}

	/**
	 * Sets the next PrivilegedSender in the token ring.
	 * 
	 * @param nextSender
	 *            The next sender in the token ring.
	 */
	public void setNextSender(SocketAddress nextSender) {
		this.tokenManager.setNextAddress(nextSender);
	}

	/**
	 * 
	 * @return The address of the next PrivilegedSender in the token ring.
	 */
	public SocketAddress getNextAddress() {
		return this.tokenManager.getNextAddress();
	}

	/**
	 * Starts the token listening functionality inside of a new thread.
	 * 
	 */
	public void start() {
		Thread tokenListener = new Thread(this.tokenManager);
		tokenListener.start();
	}

	/**
	 * Shuts down the Token listening functionality of this Sender.
	 * 
	 */
	public void shutdown() {
		this.tokenManager.shutdown();
		sender.shutdown();
	}

	/**
	 * Method to send the RPCMessage to a group of Destinations. NOTE that this
	 * method will BLOCK, barring timeout, until it receives the permission
	 * token to send the message.
	 * 
	 * @param message
	 *            The message to multicast.
	 */
	public void totalOrderBroadcast(Message message) throws IOException,
			TokenNotHeldException {

		Condition haveToken = this.tokenManager.getCondition();

		/* Aquire the lock */
		this.tokenManager.getLock().lock();
		/* While we don't have the token, wait for it */
		while (!this.tokenManager.hasToken()) {
			try {
				haveToken.await();
			} catch (InterruptedException e) {

			}
		}
		/* Re-aquire the lock here */
		Token token = this.tokenManager.getToken();
		long seqNum = token.getSeqNum();
		seqNum++;
		message.setSeqNum(seqNum);

		this.sender.send(message);

		token.setSeqNum(seqNum);
		/*
		 * Signal that the token can be safely passed on to the next
		 * PrivilegedSender in the ring
		 */
		this.tokenManager.getCondition().signal();

		this.tokenManager.getLock().unlock();
	}

}
